package com.parking.bang.rpc.core.client;

import java.net.InetSocketAddress;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.parking.bang.rpc.core.thread.RPCThreadPool;
import com.parking.bang.rpc.serialize.base.RPCSerializeProtocol;

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

public class RPCServerLoader {

	private volatile static RPCServerLoader rpcServerLoader;
	private final static String DELIMITER = ":";
	// default as JDK
	private RPCSerializeProtocol rpcSerializeProtocol = RPCSerializeProtocol.JDK;
	// default processors
	private final static int parallel = Runtime.getRuntime().availableProcessors() * 2;
	// netty thread pool
	private EventLoopGroup elg = new NioEventLoopGroup(parallel);
	private static ListeningExecutorService threadPoolExecutor = MoreExecutors
			.listeningDecorator((ThreadPoolExecutor) RPCThreadPool.getExecutor(16, -1));

	private MessageSendHandler messageSendHandler = null;

	/**
	 * connect condition
	 */
	private Lock lock = new ReentrantLock();
	private Condition connectStatus = lock.newCondition();
	private Condition handlerStatus = lock.newCondition();

	private RPCServerLoader() {
	}

	// single + synchronize
	public static RPCServerLoader getInstance() {
		if (null == rpcServerLoader) {
			synchronized (RPCServerLoader.class) {
				if (null == rpcServerLoader) {
					rpcServerLoader = new RPCServerLoader();
				}
			}
		}
		return rpcServerLoader;
	}

	// Load Server
	public void load(String serverAddress, RPCSerializeProtocol serializeProtocol) {
		String[] ipAddr = serverAddress.split(RPCServerLoader.DELIMITER);
		if (ipAddr.length == 2) {
			String host = ipAddr[0];
			int port = Integer.parseInt(ipAddr[1]);
			final InetSocketAddress remoteAddr = new InetSocketAddress(host, port);

			ListenableFuture<Boolean> listenableFuture = threadPoolExecutor
					.submit(new RPCClientInitializeTask(elg, remoteAddr, serializeProtocol));

			// future call
			Futures.addCallback(listenableFuture, new FutureCallback<Boolean>() {
				public void onSuccess(Boolean result) {
					try {
						lock.lock();

						if (messageSendHandler == null) {
							handlerStatus.await();
						}

						// Futures call, all channle on
						if (result == Boolean.TRUE && messageSendHandler != null) {
							connectStatus.signalAll();
						}
					} catch (InterruptedException ex) {
						ex.printStackTrace();
					} finally {
						lock.unlock();
					}
				}

				public void onFailure(Throwable t) {
					t.printStackTrace();
				}
			}, threadPoolExecutor);
		}
	}

	// release
	public void unLoad() {
		messageSendHandler.close();
		threadPoolExecutor.shutdown();
		elg.shutdownGracefully();
	}

	public RPCSerializeProtocol getRpcSerializeProtocol() {
		return rpcSerializeProtocol;
	}

	public void setRpcSerializeProtocol(RPCSerializeProtocol rpcSerializeProtocol) {
		this.rpcSerializeProtocol = rpcSerializeProtocol;
	}

	public MessageSendHandler getMessageSendHandler() throws InterruptedException {
		try {
			lock.lock();
			// wait till handler set
			if (messageSendHandler == null) {
				connectStatus.await();
			}
			return messageSendHandler;
		} finally {
			lock.unlock();
		}
	}

	public void setMessageSendHandler(MessageSendHandler messageSendHandler) {
		try {
			lock.lock();
			this.messageSendHandler = messageSendHandler;
			handlerStatus.signal();
		} finally {
			lock.unlock();
		}
	}

}
